package com.lzq.mq.broker.core;

import com.lzq.mq.broker.cache.CommonCache;
import com.lzq.mq.broker.model.ConsumeQueueModel;
import com.lzq.mq.broker.model.MqTopicModel;
import com.lzq.mq.broker.utils.FileNameUtil;
import com.lzq.mq.broker.utils.lock.PutMessageLock;
import com.lzq.mq.broker.utils.lock.UnfairReentrantLock;
import com.lzq.mq.common.constants.BrokerConstants;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

@Slf4j
public class ConsumeQueueMMapFileModel extends AbstractMMapFileModel {
    /**
     * topic名称
     */
    @Getter
    private String topicName;
    /**
     * 队列ID
     */
    @Getter
    private Integer queueId;
    /**
     * 添加消息的锁
     */
    private PutMessageLock putMsgLock;

    /**
     * 导入消费队列文件到MMap
     * @param topicName 主题名称
     * @param consumeQueueModel 消费队列模型
     */
    public void loadFileInMMap(String topicName, ConsumeQueueModel consumeQueueModel) {
        this.topicName = topicName;
        this.queueId = consumeQueueModel.getId();
        // 获取最新文件的绝对路径
        String filePath = getLatestConsumeQueueFile();
        try {
            doMMap(filePath, consumeQueueModel.getLastOffset(), consumeQueueModel.getOffsetLimit(), consumeQueueModel.getLatestOffset().get());
        } catch (IOException e) {
            throw new RuntimeException("映射ConsumeQueue为MMap失败");
        }
        putMsgLock = new UnfairReentrantLock();
    }

    /**
     * 获取最新的消费队列文件
     * @return 消费队列文件绝对路径
     */
    private String getLatestConsumeQueueFile() {
        MqTopicModel mqTopicModel = CommonCache.getMqTopicModelMap().get(topicName);
        if (mqTopicModel == null) {
            throw new IllegalArgumentException("Topic not exist");
        }
        List<ConsumeQueueModel> queueList = mqTopicModel.getQueueList();
        ConsumeQueueModel consumeQueueModel = queueList.get(queueId);   // queue
        if (consumeQueueModel == null) {
            throw new IllegalArgumentException("Queue not exist");
        }
        // 拿到文件
        int diff = consumeQueueModel.countDiff();
        if (diff == 0) {
            // 如果当前文件已经到达上限, 需要创建新的commitLogFile
            FilePath filePath = createNewConsumeQueueFile(consumeQueueModel.getFileName());
            return filePath.getFilePath();
        } else {
            return FileNameUtil.buildConsumeQueueFilePath(topicName, queueId, consumeQueueModel.getFileName());
        }
    }


    /**
     * 为当前MMap对应的ConsumeQueue创建新的消费队列文件
     * @param fileName 新的文件名
     * @return 新的文件路径
     */
    public FilePath createNewConsumeQueueFile(String fileName) {
        String newFileName = FileNameUtil.incrFileName(fileName);
        String newFilePath = FileNameUtil.buildConsumeQueueFilePath(topicName, queueId, newFileName);
        File newCQFile = new File(newFilePath);
        try {
            boolean flag = newCQFile.createNewFile();
            log.info("主题{}的{}号消费队列创建新存储文件:{}", topicName, queueId, newFilePath);
        } catch (IOException e) {
            log.error("主题{}的{}号消费队列创建新存储文件失败:{}", topicName, queueId, newFilePath);
            throw new RuntimeException(e);
        }
        return new FilePath(newFileName, newFilePath);
    }

    /**
     * 写入内容
     * @param content 要写入的内容
     * @param force 是否强制刷盘
     */
    public void writeContent(byte[] content, boolean force) {
        try {
            putMsgLock.lock();
            writeByteBuffer.put(content);
            if (force) {
                fileChannel.force(true);
            }
        } catch (IOException e) {
            log.error("主题{}的{}号消费队列写入内容失败", topicName, queueId);
        } finally {
            putMsgLock.unlock();
        }

    }

    /**
     * 写入内容
     * 等价于 writeContent(content, false)
     * @param content 要写入的内容
     */
    public void writeContent(byte[] content) {
        writeContent(content, false);
    }

    /**
     * 从指定偏移位置读取一个ConsumeQueueModel元素
     * @param offset 偏移量
     */
    public byte[] readContent(Integer offset) {
        return read(offset, BrokerConstants.CONSUME_QUEUE_ELEMENT_BYTE_SIZE);
    }

    /**
     * 批量读取
     * @param offset 偏移量
     * @param batch 批量读取的元素数量
     * @return 批量读取的元素
     */
    public List<byte[]> readContent(Integer offset, int batch) {
        List<byte[]> records = new ArrayList<>(batch);
        for (int i=0; i<batch; i++) {
            byte[] record = read(offset, BrokerConstants.CONSUME_QUEUE_ELEMENT_BYTE_SIZE);
            records.add(record);
            offset += BrokerConstants.CONSUME_QUEUE_ELEMENT_BYTE_SIZE;
        }
        return records;
    }
}
